package com.neusoft.bd.myflink.step05;

import com.neusoft.bd.myflink.entity.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

public class TransKeyBy {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List<WaterSensor> list = Arrays.asList(
                new WaterSensor("sensor_1", 1, 1),
                new WaterSensor("sensor_1", 2, 2),
                new WaterSensor("sensor_2", 2, 2),
                new WaterSensor("sensor_3", 3, 3)
        );
        DataStreamSource<WaterSensor> ds = env.fromCollection(list);
        KeyedStream<WaterSensor, String> waterSensorStringKeyedStream = ds.keyBy(waterSensor -> waterSensor.getId());
        waterSensorStringKeyedStream
                .sum("ts")
                .print();


        env.execute();
    }
}
